package com.example.my_mqtt.netty;

import com.example.my_mqtt.netty.handler.BootNettyMqttChannelInboundHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;


/**
 * @author CZF
 * @create 2023-02-21 15:45
 */
public class BootNettyMqttServer {

    private NioEventLoopGroup bossGroup;

    private NioEventLoopGroup workGroup;

    /**
     * 	启动服务
     */
    public void startup(int port) {

        try {
            bossGroup = new NioEventLoopGroup(1);
            workGroup = new NioEventLoopGroup();

            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workGroup);
            bootstrap.channel(NioServerSocketChannel.class);

            //设置启动参数
            bootstrap.option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .option(ChannelOption.SO_RCVBUF, 10485760);

            bootstrap.childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline channelPipeline = ch.pipeline();
                    // 设置读写空闲超时时间
                    channelPipeline.addLast(new IdleStateHandler(600, 600, 1200));
                    channelPipeline.addLast("编码器", MqttEncoder.INSTANCE);
                    channelPipeline.addLast("解码器", new MqttDecoder());
                    channelPipeline.addLast(new BootNettyMqttChannelInboundHandler());
                }
            });
            ChannelFuture f = bootstrap.bind(port).sync();
            if(f.isSuccess()){
                System.out.println("成功绑定和监听端口： " + port);
                f.channel().closeFuture().sync();
            } else {
                System.out.println("失败绑定和监听端口： " + port);
            }


//			//	绑定端口,监听服务是否启动
//			bootstrap.bind(port).addListener((ChannelFutureListener) channelFuture -> {
//				if (channelFuture.isSuccess()) {
//					System.out.println("startup success --- ");
//				} else {
//					System.out.println("startup fail ---");
//				}
//			});
        } catch (Exception e) {
            System.out.println("启动异常"+e.toString());
        }

    }

    /**
     * 	关闭服务
     */
    public void shutdown() throws InterruptedException {
        if (workGroup != null && bossGroup != null) {
            bossGroup.shutdownGracefully().sync();
            workGroup.shutdownGracefully().sync();
            System.out.println("关闭服务成功");
        }
    }

}
